/*
 * $AIST_Release: 0.9.0 $
 * Copyright 2011 Information Technology Research Institute, National
 * Institute of Advanced Industrial Science and Technology
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package hadoop

import org.apache.hadoop.io.BytesWritable
import org.apache.hadoop.mapreduce.Reducer
import aggr.Aggregators.Aggregator
import aggr.Aggregators.aggrMap
import util.PBUtils
import aggr.{SerializedTypes => ST}

abstract class SCReducerBase[T, U] extends Reducer[BytesWritable, BytesWritable, T, U] {

  type Context = Reducer[BytesWritable, BytesWritable, T, U]#Context
  var aggr: Aggregator = null

  var keyType: ST.SerializedType = null
  var valType: ST.SerializedType = null
  var nTables = 0

  override def setup(context: Context): Unit = {
    val conf = context.getConfiguration
    val c = SCConfiguration(conf)
    val protoFiles = PBUtils.convertFiles(c.protoFiles).values.toList

    val tables = c.tables
    nTables = tables.size

    if (nTables > 0) {
      val table = if (nTables == 1) {
        Some(tables.head)
      } else {
        val tableId = conf.get("sc.tableId")
        if (tableId != null) {
          tables.find { x =>
            x.id.toString == tableId
          }
        } else {
          None
        }
      }

      table match {
        case Some(t) => 
          val typeMap = ST.makeSTypeMap(protoFiles)
          keyType = typeMap(t.keyType)
          valType = typeMap(t.valType)
          val tableArgs = t.args

          val factory = aggrMap(t.aggr)

          aggr = factory.createNew(keyType, valType, t.args.map{case x:Long => x})
        case None =>
      }
    }
  }
}
